package bigdata.hbase;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseManager {
	
	private Configuration conf=null;
	private Connection conn;
	private Admin admin;
	
	public HbaseManager() {
		conf=HBaseConfiguration.create();
		conf.set("hbase.rootdir", "hdfs://localhost:9000/hbase");
		
		try {
			conn=ConnectionFactory.createConnection(conf);
			admin=conn.getAdmin();
		}catch(IOException e) {
			e.printStackTrace();
		}
	}
	
	public void close() {
		try {
			if(admin!=null) {
				admin.close();
			}
			if(null!=conn) {
				conn.close();
			}
		}catch(IOException e) {
			e.printStackTrace();
		}
	}
	
	public void createTable(String tableName,String[] colFamilys) throws IOException{
		TableName tabName=TableName.valueOf(tableName);
		if(admin.tableExists(tabName)) {
			System.out.print("table is existed ! to be deleted");
			admin.disableTable(tabName);
			admin.deleteTable(tabName);
		}
		else {
			TableDescriptorBuilder tableDescriptor=TableDescriptorBuilder.newBuilder(tabName);
			for(String str:colFamilys) {
				tableDescriptor.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(str)).build());
			}
			admin.createTable(tableDescriptor.build());
			System.out.println("Create table successfully");
		}
	}
	
	public void deleteTable(String tableName) throws IOException{
		TableName tabName=TableName.valueOf(tableName);
		if(admin.tableExists(tabName)) {
			admin.disableTable(tabName);
			admin.deleteTable(tabName);
		}
	}
	
	public void list() throws IOException{
		TableName[] tabNames=admin.listTableNames();
		for(TableName tableName:tabNames) {
			System.out.println(tableName.getNameAsString());
		}
	}
	
	public void scan(String tableName) throws IOException{
		TableName tabName=TableName.valueOf(tableName);
		if(admin.tableExists(tabName)) {
			Table table=conn.getTable(tabName);
			TableDescriptor tableDescriptor=table.getDescriptor();
			ColumnFamilyDescriptor[] columnFamilyDescriptors=tableDescriptor.getColumnFamilies();
			
			for(ColumnFamilyDescriptor cfd:columnFamilyDescriptors) {
				System.out.println(cfd.getNameAsString());					
			}
			table.close();
		}
		else {
			System.out.println(tableName + " not Existed");
		}
	}
	
	
	public void insertRow(String tableName,String rowkey, String colFamily,String col,String val) throws IOException{
		Table table=conn.getTable(TableName.valueOf(tableName));
		
		Put put=new Put(rowkey.getBytes());
		put.addColumn(colFamily.getBytes(), col.getBytes(), val.getBytes());
		table.put(put);
		table.close();
	}
	
	//to be continued

	public static void main(String[] args) throws IOException{
		// TODO Auto-generated method stub
		HbaseManager hm=new HbaseManager();
		hm.scan("student");
	}
}
